package com.sailing.lianxi.demoservice.stream;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import com.sailing.lianxi.common.Constants;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;


public class Test {

	public static void t1() {
		Properties props = new Properties();
		//指定一个应用ID，会在指定的目录下创建文件夹，里面存放.lock文件
		props.put("zookeeper.connect", Constants.KAFKA_ZOOKEEPER_URL);
		//指定一个路径创建改应用ID所属的文件
		props.put("group.id", "mygroup002");
		props.put("zookeeper.session.timeout", "4000");
		props.put("zookeeper.sync.time.ms", "2000");
		//自动提交偏移量的间隔
		props.put("auto.commit.interval.ms", "1000");
//		props.put("auto.offset.reset", "earliest");
		ConsumerConfig config = new ConsumerConfig(props);
		//设置每个topic的线程数
		Map<String, Integer> topicMap = new HashMap<String, Integer>();
		topicMap.put(Constants.TOPIC_NAME, 4);
		 ConsumerConnector consumer = Consumer.createJavaConsumerConnector(config);
		 Map<String, List<KafkaStream<byte[], byte[]>>>  consumerMap = consumer.createMessageStreams(topicMap);
		 List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(Constants.TOPIC_NAME);
		 for (KafkaStream<byte[], byte[]> kafkaStream : streams) {
			 ConsumerIterator<byte[], byte[]> iterator = kafkaStream.iterator();
			 while(iterator.hasNext()) {
				 System.out.println("message:"+new String(iterator.next().message()));
			 }
		}
	}
	
	public static void main(String[] args) {
		t1();
	}
}
